set spark.sql.crossJoin.enabled=true;
with yl_oms_oms_waybill_cust as ( --客户发件量
select
      dt as date_time --日期
     ,nvl(customer_code,'其他') as customer_code  --客户Code，后续关联需要给值
     ,sum(total_bill_count) as  total_bill_count --总票数
     ,sum(total_weight_count) as total_weight_count --总重量（内部计费总量）
     from jms_dm.dm_customer_province_bill_amount_dt 
     where dt between date_add(trunc(add_months(date_add('{{ execution_date  | cst_ds }}',-60),-3),'MM'),-1) and '{{ execution_date  | cst_ds }}'
     group BY
      dt       --日期
     ,customer_code  --客户Code
)

insert overwrite table jms_dm.dm_contract_area_bill_customer_detail_dt
select 
 date_time        --日期
,if(customer_code ='其他',null,customer_code)  as customer_code --客户来源
,customer_name    --客户名称
,total_bill_count  --当日件量
,total_weight_count --当日重量
,total_bill_mm   --当月件量
,total_days_mm --当月发货天数
,avg_total_bill   --日均发货件量
,up_bill_cnt      --上一日发件量
,up_weight_cnt     --上一日发件重量
,customer_level   
,staff_code   --业务员code
,staff_name   --业务员name
,network_code --网点code
,network_name --网点name
,belong_type  --所属类别'所属类别1 直营公司2 第三方3 底价业务员',
,source_id   --客户来源id
,agent_code  --网点所属代理区code
,agent_name  --网点所属代理区name
,fran_code   --网点所属加盟商code
,fran_name   --网点所属加盟商name
,null as total_days_mm_bit  --没用的字段
,customer_create_time
,total_weight_mm --当月总重量
,dt
from 
(
    select 
     main.date_time        --日期
    ,main.customer_code    --客户来源
    ,cust.name as customer_name    --客户名称
    ,main.total_bill_count  --当日件量
    ,main.total_weight_count --当日重量
    ,main.total_bill_mm   --当月件量
    ,size(main.total_days_mm) as total_days_mm --当月发货天数
    ,cast(main.total_bill_mm/size(main.total_days_mm) as decimal(16,0)) as avg_total_bill   --日均发货件量
    ,lag(main.total_bill_count,1,0) over(partition by main.customer_code order by main.date_time) as up_bill_cnt --上一日发件量
    ,lag(main.total_weight_count,1,0) over(partition by main.customer_code order by main.date_time) as up_weight_cnt --上一日发件重量
    ,case when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>5000 then 'A'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=5000 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>3000 then 'B+'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=3000 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>1000  then 'B'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=1000 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>200  then 'C'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=200 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>50    then 'D'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=50 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>30    then 'E'
          when cast(total_bill_mm/size(total_days_mm) as decimal(16,0))<=30 and cast(total_bill_mm/size(total_days_mm) as decimal(16,0))>0      then 'F'
         end as  customer_level   
    ,cust.market_staff_code as staff_code   --业务员code
    ,staff2.name as staff_name   --业务员name
    ,net.code as network_code --网点code
    ,net.name as network_name --网点name
    ,nvl(staff.belong_type,-1) as belong_type  --所属类别'所属类别1 直营公司2 第三方3 底价业务员',
    ,if(cust.source_id in ('326','348','352','353','328'),source_id,'其他') as source_id--客户来源id
    ,net.agent_code  --网点所属代理区code
    ,net.agent_name  --网点所属代理区name
    ,net.fran_code   --网点所属加盟商code
    ,net.fran_name   --网点所属加盟商name
    ,cust.create_time as customer_create_time
    ,main.total_weight_mm
    ,main.date_time as dt
    from 
    (
        select
         dt.date_mid_desc  as date_time
        ,cust1.customer_code
        ,nvl(way.total_bill_count,0) as total_bill_count  --当日出件量
        ,nvl(way.total_weight_count,0) as total_weight_count --当日出件重量
        ,sum(way.total_bill_count) over(partition by date_format(dt.date_mid_desc,'yyyy-MM'),cust1.customer_code order by dt.date_mid_desc) as total_bill_mm        --当月发货件量
        ,sum(way.total_weight_count) over(partition by date_format(dt.date_mid_desc,'yyyy-MM'),cust1.customer_code order by dt.date_mid_desc) as total_weight_mm    --当月发货总重量
        ,collect_set(way.date_time) over(partition by date_format(dt.date_mid_desc,'yyyy-MM'),cust1.customer_code order by  dt.date_mid_desc ) as total_days_mm     --当月发货天数
        from 
        (select customer_code from yl_oms_oms_waybill_cust group by customer_code )cust1 
        inner join jms_dim.dim_date_info_base dt on dt.date_mid_desc between date_add(trunc(add_months(date_add('{{ execution_date  | cst_ds }}',-60),-3),'MM'),-1) and '{{ execution_date  | cst_ds }}' and 1=1
        left join yl_oms_oms_waybill_cust way on dt.date_mid_desc=way.date_time and cust1.customer_code=way.customer_code
    )main
    left join jms_dim.dim_lmdm_sys_customer cust on main.customer_code=cust.code and cust.is_enable=1 and cust.is_delete=1
    left join jms_dim.dim_yl_nwm_cost_config staff on staff.config_type_code=3 and main.date_time between staff.start_time and staff.end_time and staff.code=cust.market_staff_code
    left join jms_dim.dim_lmdm_sys_staff staff2 on staff2.code=cust.market_staff_code
    left join jms_dim.dim_network_whole_massage net on cust.network_id=net.id
) tt where dt between date_add('{{ execution_date | cst_ds }}',-60) and '{{ execution_date  | cst_ds }}'
distribute by dt, abs(hash(customer_code)) % 2 ;